package ckR

import (
	"db2s/global"
	"fmt"
	"reflect"
	"strings"
	"sync/atomic"
)

func SendMsgExecFunc(Parameter *int64, delta int64) ExecFuncS {
	return ExecFuncS{
		CallFunc: func(s any, d int64) {
			switch fmt.Sprintf("%v", reflect.TypeOf(s)) {
			case "*int64":
				atomic.AddInt64(s.(*int64), d)
			}
		},
		Parameter: Parameter,
		Delta:     delta,
	}
}
func SendMsgEndExecFunc(Parameter []*int64, delta int64) ExecFuncS {
	return ExecFuncS{
		CallFunc: func(s any, d int64) {
			switch fmt.Sprintf("%v", reflect.TypeOf(s)) {
			case "[]*int64":
				for _, v := range s.([]*int64) {
					atomic.AddInt64(v, d)
				}
			}
		},
		Parameter: Parameter,
		Delta:     delta,
	}
}
func SendMsgExecSumFunc(param any) global.EFunc {
	return global.EFunc{
		FuncName: func(event string,s any) (r any) {
			var point []any
			var val int64
			switch s.(type) {
			case []any:
				for k, v := range s.([]any) {
					switch k {
					case 0:
						switch reflect.TypeOf(v).String() {
						case "interface {}", "*int64":
							point = append(point, v)
						case "[]interface {}":
							for _, q := range v.([]any) {
								point = append(point, q)
							}
						}
					case 1:
						val = v.(int64)
					}
				}
			}
			for _, v := range point {
				switch v.(type) {
				case *int64:
					atomic.AddInt64(v.(*int64), val)
				}
			}
			return
		},
		Params: param,
	}
}

func SendMsgActionFunc(param any) global.EFunc {
	return global.EFunc{
		FuncName: func(event string,s any) (r any) {
			var check CheckMode
			var topic Topic
			var outFay string
			switch s.(type) {
			case map[string]any:
				if v, ok := param.(map[string]any)["outFay"]; ok {
					outFay = v.(string)
				}
				if v, ok := param.(map[string]any)["check"]; ok {
					check = v.(CheckMode)
				}
				if v, ok := param.(map[string]any)["topic"]; ok {
					topic = v.(Topic)
				}
			}
			PlanContext.topicEndAdd(check, topic)
			if strings.EqualFold(outFay, "bar") {
				PlanContext.taskBarEndTopicAdd(param)
			}
			return
		},
		Params: param,
	}
}

func insertValueSumTotal(insertSuffix any) (insertSum int64, insertSql any) {
	if insertSuffix == nil {
		return
	}
	switch fmt.Sprintf("%v", reflect.TypeOf(insertSuffix)) {
	case "map[int64]interface {}":
		for k, v := range insertSuffix.(map[int64]any) {
			insertSum = k
			insertSql = v
		}
	case "map[int64]*string":
		for k, v := range insertSuffix.(map[int64]*string) {
			insertSum = k
			insertSql = v
		}
	case "map[int64][]*string":
		for k, v := range insertSuffix.(map[int64][]*string) {
			insertSum = k
			insertSql = v
		}
	case "map[int64][][]*string":
		for k, v := range insertSuffix.(map[int64][][]*string) {
			insertSum = k
			insertSql = v
		}
	}
	return
}
func getRowsSum(tt any) (sum int64) {
	switch tt.(type) {
	case int64:
		sum = tt.(int64)
	case []*string:
		sum = int64(len(tt.([]*string)))
	case [][]*string:
		sum = int64(len(tt.([][]*string)))
	}
	return
}

func barTerminalRowsUpdate(tt any, left string, plan *SchedulePlan2) {
	if GlobalPConfigs.result.Teletypewriter != "bar" || PlanContext.TaskBarSubsStatusGet(plan.subTaskInfo) == -2 {
		return
	}
	switch left {
	case "left":
		plan.MQ.Speed.ReadSumAdd(getRowsSum(tt))
		atomic.AddInt64(plan.TPod.Sync.SelectRows, getRowsSum(tt))
		PlanContext.SubTaskBarAccumulate(map[string]any{"subTask": plan.subTaskInfo, "select": getRowsSum(tt), "insert": int64(0)})
	case "right":
		plan.MQ.Speed.WriteSumAdd(getRowsSum(tt))
		atomic.AddInt64(plan.TPod.Sync.InsertRows, getRowsSum(tt))
		PlanContext.SubTaskBarAccumulate(map[string]any{"subTask": plan.subTaskInfo, "select": int64(0), "insert": getRowsSum(tt)})
	}
}

func getFirstMsgFinishStatus(plan *SchedulePlan2) bool {
	return atomic.LoadInt64(plan.MQ.FirstMsg.ProductSendMsgFinishMark) == 1
}
func getSecondMsgFinishStatus(plan *SchedulePlan2) bool {
	return atomic.LoadInt64(plan.MQ.FirstMsg.ProductMsg) == atomic.LoadInt64(plan.MQ.SecondMsg.ProductSendMsgFinishMark)
}
func getFirstMsgProductCustomerEq(plan *SchedulePlan2) bool {
	return atomic.LoadInt64(plan.MQ.FirstMsg.ProductMsg) == atomic.LoadInt64(plan.MQ.FirstMsg.CustomerMsg)
}
func getSecondMsgProductCustomerEq(plan *SchedulePlan2) bool {
	return atomic.LoadInt64(plan.MQ.SecondMsg.ProductMsg) == atomic.LoadInt64(plan.MQ.SecondMsg.CustomerMsg)
}
func getFirstObjectProductCustomerEq(plan *SchedulePlan2) bool {
	return atomic.LoadInt64(plan.MQ.FirstMsg.ProductObject) == atomic.LoadInt64(plan.MQ.FirstMsg.CustomerObject)
}
func abnormalErrorQuit(plan *SchedulePlan2) bool {
	return atomic.LoadInt64(plan.Status.ErrEnd) < 0
}

func terminalOutputTPod(plan *SchedulePlan2) {
	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
		plan.TPod.Result = "error"
	}
	plan.TPod.Seq = plan.Object.TaskSeq
	newTer.Write(plan.TPod)
	plan.XlsTerminal.Data = *plan.TPod
	newCell.Write(plan.XlsTerminal)
}

func mqStatusMonitorAbnormalQuit(event string,i any) (r any) {
	var plan *SchedulePlan2
	defer func() {
		terminalOutputTPod(plan)
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "error")
		baseMsgSendBody.MsgCurryDone()
		baseMsgSendBody.MsgCurryWgDone()
	}()
	switch i.(type) {
	case *SchedulePlan2:
		plan = i.(*SchedulePlan2)
	default:
		return
	}
	//_ = singleTableFinishAnalyzeAction(plan)
	//_ = singleTableFinishXlsWriteAction(plan)
	return
}
func generalObjectConfig(plan *SchedulePlan2) string {
	var schema, table string
	var options = []string{" "}
	schema = fmt.Sprintf("%v:%v", plan.Object.Schema, plan.Object.MSchema)
	table = fmt.Sprintf("%v:%v", plan.Object.Table, plan.Object.MTable)
	if len(plan.TableAttributes.ForceIndex) > 0 {
		options = append(options, fmt.Sprintf(" forceIndex:\"%v\" ", plan.TableAttributes.ForceIndex))
	}
	if len(plan.TableAttributes.ForcePart) > 0 {
		options = append(options, fmt.Sprintf(" forcePart:\"%v\" ", plan.TableAttributes.ForcePart))
	}
	if len(plan.TableAttributes.WhereSql) > 0 {
		options = append(options, fmt.Sprintf(" whereSql:\"%v\" ", plan.TableAttributes.WhereSql))
	}
	if len(plan.TableAttributes.WhereAdd) > 0 {
		options = append(options, fmt.Sprintf(" whereAdd:\"%v\" ", plan.TableAttributes.WhereAdd))
	}
	if len(plan.TableAttributes.PX) > 0 {
		options = append(options, fmt.Sprintf(" px:%v ", plan.TableAttributes.PX))
	}
	return fmt.Sprintf("\"%v\" { \"%v\" {role:\"%v\" %v }}\n", schema, table, plan.TableAttributes.Role, strings.Join(options, ","))
}
func mqStatusMonitorNormalQuit(event string,i any) (r any) {
	//var eventNew = "[mqStatusMonitorNormalQuit]"
	var plan *SchedulePlan2
	defer func() {
		terminalOutputTPod(plan)
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "finish")
		baseMsgSendBody.MsgCurryDone()
		baseMsgSendBody.MsgCurryWgDone()
		infoWriteInfo(fmt.Sprintf("(%v) %v Table %v.%v data migration is completed.", plan.Object.ObjectLogSeq, event, plan.Object.Schema, plan.Object.Table))
	}()
	switch i.(type) {
	case *SchedulePlan2:
		plan = i.(*SchedulePlan2)
	default:
		return
	}
	debugWriteInfo(fmt.Sprintf("(%d) %v It is detected that the message queue production quantity and consumption quantity in table %v.%v are equal.", plan.Object.ObjectLogSeq, event, plan.Object.Schema, plan.Object.Table))
	//_ = singleTableFinishAnalyzeAction(plan)
	//_ = singleTableFinishXlsWriteAction(plan)
	if atomic.LoadInt64(plan.TPod.Rows.SourceNumber) != atomic.LoadInt64(plan.TPod.Rows.DestNumber) {
		plan.TPod.Result = "abnormal"
		if err := newTextVerifyAbnormal.Write("", generalObjectConfig(plan)); err != nil {
			fmt.Println("-------err:")
		}
	}
	return
}
func getMsgSendData(plan *SchedulePlan2, si map[string]any) (res map[string]any) {
	res = map[string]any{
		"plan":    plan,
		"check":   getCheckMod(),
		"subTask": plan.subTaskInfo,
		"err":     plan.Status.ErrEnd,
		"outFay":  GlobalPConfigs.result.Teletypewriter,
	}
	for k, v := range si {
		res[k] = v
	}
	return
}
func execSecondMsgCustomerMqSum(s *int64) {
	atomic.AddInt64(s, 1)
}
